package org.kth.dks.dks_dht;

import java.net.MalformedURLException;

import org.apache.log4j.Logger;
import org.kth.dks.DKSAppInterface;
import org.kth.dks.DKSDHTVisualizationInterface;
import org.kth.dks.DKSMCastDHTInterface;
import org.kth.dks.DKSObject;
import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_exceptions.DKSIdentifierAlreadyTaken;
import org.kth.dks.dks_exceptions.DKSMCastNoSouchGroup;
import org.kth.dks.dks_exceptions.DKSNodeAlreadyRegistered;
import org.kth.dks.dks_exceptions.DKSRefNoResponse;
import org.kth.dks.dks_exceptions.DKSTooManyRestartJoins;
import org.kth.dks.dks_marshal.DKSMarshal;
import org.kth.dks.dks_marshal.DKSMessage;
import org.kth.dks.util.AsyncOperation;

public class DKSMCastDHTImpl
    implements DKSMCastDHTInterface, DKSDHTCallback, DKSDHTVisualizationInterface {

	private Logger log = Logger.getLogger(DKSMCastDHTImpl.class);
    DKSDHTImpl dht;
    DKSDHTCallback callbacks;

    AsyncOperation createOp = null; // Used to handle create group
				    // operations

    AsyncOperation infoOp = null; // Used to handle group info
				  // operations

    /**
     * Create a new multicast group with a specific group
     * id. Automatically become a member of the group if the group
     * does not exists. If the group already exists
     * DKSNodeAlreadyRegistered will be thrown.
     *
     */
    public DKSMCastDHTInterface createGroup(long gid,
					    DKSDHTCallback cb)
	throws DKSIdentifierAlreadyTaken,
	       DKSTooManyRestartJoins {

	/* Figure the new node's identity */
	DKSRef ref = dht.myDKSImpl.getDKSRef();
	DKSOverlayAddress ov = new DKSOverlayAddress(ref.getID(),
						     gid,
						     ref.getGUID());

	DKSRef grpRef = DKSRef.newDKSRefByParts(ov, ref.getDKSNetAddress());

	/* Create the new node */
	DKSMCastDHTImpl newNode = new DKSMCastDHTImpl(dht.myDKSImpl.getCM(),
						      ov,
						      cb);

	DKSMessage msg = new CreateMCastGroupMsg(gid,
						 grpRef,
						 newNode.getDHT().myDKSImpl.getDKSRef());

	log.debug(
			 "createGroup(" + gid + ", " +
			 grpRef + ") message sent");

	dht.routeAsync(gid, dht.message2obj(msg));

	createOp = AsyncOperation.start();

	log.debug(
			 "createGroup(" + gid +
			 ") waiting for reply");

	CreateMCastGroupReplyMsg reply;
	try {
	    reply =
		(CreateMCastGroupReplyMsg) createOp.waitOn();
	    createOp = null;
	} catch(Exception ex) {
	    newNode.unregisterNode();
	    throw new DKSIdentifierAlreadyTaken();
	}

	log.debug(
			 "createGroup(" + gid + ") got reply -> " +
			 (reply.getStatus() ? "true" : "false"));
	if(!reply.getStatus()) {
	    newNode.unregisterNode();
	    throw new DKSIdentifierAlreadyTaken();
	}

	log.debug(
			 "createGroup(" + gid +
			 ") new node created");

	newNode.create();

	log.debug(
			 "createGroup(" + gid +
			 ") completed successfully");

	return newNode;
    }

    /**
     * Join an existing multicast group with a specific group id. If
     * the group does not exist DKSMCastNoSouchGroup is thrown.
     *
     */
    public DKSMCastDHTInterface joinGroup(long gid, DKSDHTCallback cb)
	throws DKSMCastNoSouchGroup,
	       DKSIdentifierAlreadyTaken,
	       DKSRefNoResponse {
	DKSRef ref = dht.myDKSImpl.getDKSRef();

	DKSOverlayAddress ov = new DKSOverlayAddress(ref.getID(),
						     gid,
						     ref.getGUID());

	DKSRef grpRef = DKSRef.newDKSRefByParts(ov, ref.getDKSNetAddress());
	DKSMCastDHTImpl newNode = new DKSMCastDHTImpl(dht.myDKSImpl.getCM(),
						      ov,
						      cb);

	log.debug(
			 "joinGroup(" + gid + ") new node created");

	DKSRef[] refs = null;

	try {
	    refs = getGroupInfo(gid);
	} catch(DKSMCastNoSouchGroup e) {
	    newNode.unregisterNode();
	    throw e;
	}

	// Loop until we find a working node
	for(int i = 0; i < refs.length; i++) {
	    log.debug(
			     "Trying to connect to " + refs[i].toString());
	    try {
		newNode.join(refs[i]);

		log.debug(
				 "joinGroup(" + gid +
				 ") new node started, returning");

		// XXXX Tell the responsible node that we have joined
		dht.routeAsync(gid,
			       dht.message2obj(new JoinMCastGroupMsg(gid,
								     newNode.getDHT().myDKSImpl.getDKSRef())));

		return newNode;
	    } catch(DKSTooManyRestartJoins e) {
		// Try the next node
	    }
	}
	// We failed, unregister the node
	newNode.unregisterNode();

	log.debug(
			 "Failed to join group(" + gid + ")");
	throw new DKSRefNoResponse(refs[0].getDKSNetAddress());
    };

    /**
     * Return the members of the group.
     *
     */
    public DKSRef[] getGroupInfo(long gid) throws DKSMCastNoSouchGroup {


	DKSMessage msg =
	    new MCastGroupInfoMsg(gid, dht.myDKSImpl.getDKSRef());

	log.debug(
			 "groupinfo(" + gid + ") message sent");
	dht.routeAsync(gid, dht.message2obj(msg));

	infoOp = AsyncOperation.start();

	log.debug(
			 "groupinfo(" + gid + ") waiting for reply");

	MCastGroupInfoReplyMsg reply;

	try {
	    reply =
		(MCastGroupInfoReplyMsg) infoOp.waitOn();
	} catch(Exception ex) {
	    infoOp = null;
	    throw new DKSMCastNoSouchGroup();
	}

	log.debug(
			 "groupinfo(" + gid + ") got reply -> " +
			 (reply.getRefs().length == 0 ? "no souch group" : "success"));
	if(reply.getRefs().length == 0)
	    throw new DKSMCastNoSouchGroup();
	return reply.getRefs();
    }


    /**
     * Value is added to the list of bindings associated with key.
     *
     */
    public void addToBinding(long key, DKSObject value) {
	dht.addToBinding(key, addMeta(false, value));
    }

    /**
     * Value is removed from the list of bindings associated with key.
     *
     * Any ocurrence of value in the list is determined by comparison
     * of the DKSObject content identifier.
     */
    public void removeFromBinding(long key, DKSObject value) {
	dht.removeFromBinding(key, addMeta(false, value));
    }

    /**
     * All ocurrences of oldValue in the bindings associated with key
     * are replaced by newValue.
     *
     * Any ocurrence of value in the list is determined by comparison
     * of the DKSObject content identifier.
     */
    public void changeBinding(long key,
			      DKSObject oldValue,
			      DKSObject newValue) {
	dht.changeBinding(key,
			  addMeta(false, oldValue),
			  addMeta(false, newValue));
    }

    /**
     * Returns a vector of DKSObjects associated with <code>key</code>
     * in the DHT
     * @param key long specifying the key to be looked up
     * @return <code>DKSObject[]</code> with values associated to
     * <code>key</code>
     * @return <code>null</code> if no values are associated with
     * <code>key</code>
     */
    public DKSObject[] lookupBinding(long key) {
	return removeMeta(dht.lookupBinding(key));
    }

    /**
     * Returns an array of all bindings associated with keys greater
     * than or equal to minKey, and less than or equal to maxKey.
     */
    public DKSObject[] lookupBinding(long minKey, long maxKey) {
	return removeMeta(dht.lookupBinding(minKey, maxKey));
    }

    /**
     * Broadcast a message to all nodes
     */
    public void	broadcast(DKSObject message) {
	dht.broadcast(message);
    }

    /**
     * Broadcasts to nodes with identifiers within the rand startId and endId []
     * @param message DKSObject message to be sent
     * @param startId long beginning of the interval
     * @param endId long end of the interval
     */
    public void broadcastRestricted(DKSObject message, long startId, long endId) {
      dht.broadcastRestricted(message, startId, endId);
    }


    
    /**
     * Adds a new node to the network managed by the local DKS
     * implementation.
     *
     * @param nodeId The unique DKS node identifier (key) for the new node.
     * @param nodeAddress The location of the new node on the physical
     * underlying network.
     */
    public void join(DKSRef existingnodeAddress)
	throws DKSIdentifierAlreadyTaken, DKSTooManyRestartJoins, DKSRefNoResponse
    {
	dht.join(existingnodeAddress);
    }

    /**
     * Called by the first node in a ring, this will create a new DKS
     * network containing one node only
     * @param existingnodeAddress DKSOverlayAddress
     */
    public void create() {
	dht.create();
    }

    /**
     * Set the logging level
     * @param level int, 0 means no logging, 1 means debug prints, 2
     * means XML dumps
     */
    public void logLevel(int level) {
	dht.logLevel(level);
    }

    /**
     * Disconnect the local node from the network.
     */
    public void leave() {
	dht.leave();
	dht.unregisterNode();
    }

    /**
     * Will find the DKS node that is responsible for the given
     * identifier
     * @param identifier whose responsible node is being queried for
     * @return DKSRef reference to the responsible node
     */
    public DKSRef findResponsible(long identifier) {
	return dht.findResponsible(identifier);
    }

    /**
     * Will route a message the DKS node that is responsible for the
     * given identifier, and call routeCallback() at that node and
     * give it the payload, upon which the responsible node will
     * respond with a payload which is returned to the caller
     * @param identifier whose responsible node being routed to
     * @param payload to be sent to the responsible node
     * @return DKSOBject payload sent back by the responsible node
     * @see DKSAppInterface
     */
    public DKSObject route(long identifier, DKSObject payload) {
	return dht.route(identifier, payload);
    }

    /**
     * Will route a message the DKS node that is responsible for the
     * given identifier, and call routeCallbackAsync() at that node
     * and give it the payload, upon which the responsible node will
     * respond with a payload which is returned to the caller
     * @param identifier whose responsible node being routed to
     * @param payload to be sent to the responsible node
     * @see DKSAppInterface
     */
    public void routeAsync(long identifier, DKSObject payload) {
        dht.routeAsync(identifier, payload);
    }

    /**
     * Will route a message from fromNode to the DKS node that is responsible for the
     * given identifier, and call routeCallbackAsync() at that node
     * and give it the payload, upon which the responsible node will
     * respond with a payload which is returned to the caller
     * @param identifier whose responsible node being routed to
     * @param payload to be sent to the responsible node
     * @see DKSAppInterface
     */
    public void routeAsyncFrom(long identifier, DKSObject payload, DKSRef fromNode) {
        dht.routeAsyncFrom(identifier, payload, fromNode);
    }

    /**
     * Returns the DKSURL (stringified DKSRef) used by this DKS node
     * @return String
     * @see org.kth.dks.dks_comm.DKSRef
     */
    public String getDKSURL() {
	return dht.getDKSURL();
    }

    /**
     * Every object implementing this interface gets registered in a
     * ConnectionManager, this method this method unregisters the
     * object in the ConnectionManager. This method needs to be called
     * especially if another object of this type is to be instantiated
     * with the same DKSNetAddress (overlay identity)
     */
    public void unregisterNode() {
	dht.unregisterNode();
    }

    /**
     * Constructor
     */
    /**
     * cb is the callback object
     */
    public DKSMCastDHTImpl(ConnectionManager cm,
			   DKSOverlayAddress over,
			   DKSDHTCallback cb)
    {
	log.debug( "--------------- 0 " + over);
	try {
	    dht = new DKSDHTImpl(cm, over, this);
	} catch(DKSNodeAlreadyRegistered e) {
	    log.error( "internal error " + e);
	}
	log.debug( "--------------- 1");

	callbacks = cb;
	DKSMessage.addMessageTypePrefixed(JoinMCastGroupMsg.NAME,
                                          "dks_dht.JoinMCastGroupMsg");
        DKSMessage.addMessageTypePrefixed(CreateMCastGroupMsg.NAME,
                                          "dks_dht.CreateMCastGroupMsg");
        DKSMessage.addMessageTypePrefixed(CreateMCastGroupReplyMsg.NAME,
                                          "dks_dht.CreateMCastGroupReplyMsg");

        DKSMessage.addMessageTypePrefixed(MCastGroupInfoMsg.NAME,
                                          "dks_dht.MCastGroupInfoMsg");
        DKSMessage.addMessageTypePrefixed(MCastGroupInfoReplyMsg.NAME,
                                          "dks_dht.MCastGroupInfoReplyMsg");

    }

    /**
     * Receive a broadcast message
     */
    public void dhtBroadcastCallback(DKSObject value) {
	callbacks.dhtBroadcastCallback(value);
    }

    /**
     * The callback is called at the responsible node when a route()
     * message reaches its destination and the message is not handled
     * at a lower abstraction layer. It receives a DKSMessage, and
     * returns a possibly new DKSMessage <b>INVARIANT</b>: the
     * implementor of this method SHOULD immediately return, as
     * otherwise the underlying DKSNode cannot process new messagesl,
     * and consequently the method cannot recursively call other
     * methods in the DKSInterface as the DKSNode will not be able to
     * receive response messages. Consider using routeAsync() in
     * DKSInterface.
     * @param identifier long, the identifier searched for in route,
     * maybe not interesting for the app in many cases.
     * @param payload DKSObject, the actual payload sent by the source
     * @return DKSObject the payload to be routed back or null if the
     * message is unhandled
     */
    public DKSMessage dhtRouteCallback(long identifier, DKSMessage value) {
	log.debug( "dhtRouteCallback");
	return callbacks.dhtRouteCallback(identifier, value);
    }


    /**
     * The callback is called at the responsible node when a route()
     * message reaches its destination and the message is not handled
     * at a lower abstraction layer. It receives a DKSMessage
     * @param identifier long, the identifier searched for in route,
     * maybe not interesting for the app in many cases.
     * @param payload DKSObject, the actual payload sent by the source
     */
    public void dhtRouteCallbackAsync(long identifier, DKSMessage payload) {
	log.debug( "dhtRouteCallbackAsync");
	if(payload instanceof CreateMCastGroupMsg) {
	    log.debug(
			     "dhtRouteCallbackAsync: got CreateMCastGroupMsg");
	    CreateMCastGroupMsg msg = (CreateMCastGroupMsg)payload;
	    handleCreateGroup(msg);
	} else if(payload instanceof CreateMCastGroupReplyMsg) {
	    log.debug(
			     "dhtRouteCallbackAsync: got CreateMCastGroupReplyMsg");
	    handleCreateGroupReply((CreateMCastGroupReplyMsg)payload);
	} else if(payload instanceof JoinMCastGroupMsg) {
	    log.debug(
			     "dhtRouteCallbackAsync: got JoinMCastGroupMsg");
	    handleJoinGroup((JoinMCastGroupMsg)payload);
	} else if(payload instanceof MCastGroupInfoMsg) {
	    log.debug(
			     "dhtRouteCallbackAsync: got MCastGroupInfoMsg");
	    handleGroupInfo((MCastGroupInfoMsg)payload);
	} else if(payload instanceof MCastGroupInfoReplyMsg) {
	    log.debug(
			     "dhtRouteCallbackAsync: got MCastGroupInfoReplyMsg");
	    handleGroupInfoReply((MCastGroupInfoReplyMsg)payload);
	} else
	    callbacks.dhtRouteCallbackAsync(identifier, payload);
    }

    /**
     * Return a reference to the underlying DKSDHT
     */
    public DKSDHTImpl getDHT() {
	return dht;
    }


    /**
     * Add meta classification to a DKS object
     */
    DKSObject addMeta(boolean isMeta, DKSObject value) {
	byte[] in = value.getData();
	byte[] data = new byte[in.length + 1];
	data[0] = isMeta ? (byte)1 : (byte)0;
	for(int i = 0; i < in.length; i++)
	    data[i + 1] = in[i];
	return new DKSObject(data);
    }

    /**
     * Return true if the value contains meta information
     */
    boolean isMeta(DKSObject value) {
	return value.getData()[0] == 1;
    }

    /**
     * Remove the meta classification from a DKS object
     */
    DKSObject removeMeta(DKSObject value) {
	byte[] in = value.getData();
	byte[] data = new byte[in.length - 1];

	for(int i = 0; i < (in.length - 1); i++)
	    data[i] = in[i + 1];
	return new DKSObject(data);
    }

    /**
     * Remove the meta classification from an array of DKS objects,
     * only return non meta objects
     */
    DKSObject[] removeMeta(DKSObject[] value) {
	int n = 0;
	for(int i = 0; i < value.length; i++)
	    if(!isMeta(value[i]))
		n++;

	DKSObject[] result = new DKSObject[n];
	for(int i = 0, j = 0; i < value.length; i++)
	    if(!isMeta(value[i]))
		result[j++] = removeMeta(value[i]);

	return result;
    }

    /**
     * Remove the meta classification from an array of DKS objects,
     * only return meta objects
     */
    DKSObject[] getMeta(DKSObject[] value) {
	int n = 0;

	for(int i = 0; i < value.length; i++)
	    if(isMeta(value[i]))
		n++;

	DKSObject[] result = new DKSObject[n];
	for(int i = 0, j = 0; i < value.length; i++)
	    if(isMeta(value[i]))
		result[j++] = removeMeta(value[i]);

	return result;
    }

    /**
     *
     * Handle the reception of a CreateMCastGroupMsg message.
     */
    void handleCreateGroup(CreateMCastGroupMsg msg) {
	log.debug(
			 "Got create group for " +
			 msg.getGroupId() + " from " + msg.getSource() +
			 " participant: " + msg.getGroupNode());
	// Do lookup
	DKSObject[] nodes = getMeta(dht.lookupBinding(msg.getGroupId()));

	dht.routeAsync(msg.getSource().getID(),
		       dht.message2obj(new CreateMCastGroupReplyMsg(nodes.length == 0)));
	if(nodes.length != 0)
	    return;

	// Now add the node to the bindings for this group
	// XXXX How should duplicates be handled?
	dht.addToBinding(msg.getGroupId(),
			 addMeta(true,
				 new DKSObject(msg.getGroupNode().toString().getBytes())));
    }

    /**
     * Handle the reception of a CreateMCastGroupReplyMsg message.
     */
    void handleCreateGroupReply(CreateMCastGroupReplyMsg msg) {
	if(createOp != null) {
	    log.debug(
			     "dhtRouteCallbackAsync: completing future");
	    createOp.complete(msg);
	}
    }

    /**
     *
     * Handle the reception of a MCastGroupInfoMsg message.
     */
    void handleJoinGroup(JoinMCastGroupMsg msg) {
	DKSRef participant = msg.getGroupNode();
	log.debug(
			 "Got join group for " +
			 msg.getGroupId() + " from " + participant);

	dht.addToBinding(msg.getGroupId(),
			 addMeta(true,
				 new DKSObject(participant.toString().getBytes())));
    }

    /**
     *
     * Handle the reception of a MCastGroupInfoMsg message.
     */
    void handleGroupInfo(MCastGroupInfoMsg msg) {
	log.debug(
			 "Got group info request for " +
			 msg.getGroupId() + " from " + msg.getSource());
	// Do lookup
	DKSObject[] nodes = getMeta(dht.lookupBinding(msg.getGroupId()));
	DKSRef[] refs = new DKSRef[nodes.length];

	log.debug(
			 nodes.length + " known nodes");

	for(int i = 0; i < nodes.length; i++) {
	    try {
		String str = new String(nodes[i].getData());
		refs[i] = DKSRef.valueOf(str);
		log.debug( refs[i].toString());
	    } catch(MalformedURLException e) {
		// XXXX Abort
		log.error( "Bad meta data");
	    }
	}

	MCastGroupInfoReplyMsg reply = new MCastGroupInfoReplyMsg();
	reply.addRefs(refs);

	dht.routeAsync(msg.getSource().getID(), dht.message2obj(reply));
    }

    public void send(DKSRef target, DKSMessage message) {
    	dht.send(target, message);
    }
    
    /**
     * Handle the reception of a MCastGroupInfoReplyMsg message.
     */
    void handleGroupInfoReply(MCastGroupInfoReplyMsg msg) {
	if(infoOp != null) {
	    log.debug(
			     "dhtRouteCallbackAsync: completing future");
	    infoOp.complete(msg);
	}
    }

    public DKSRef getDKSRef() {
    	return dht.getDKSRef();
    }

    /**
     * @see DKSInterface.addMsgHandler()
     */
    public boolean addMsgHandler(DKSMessage msg, Object handlerObject, String methodName) {
    	return dht.addMsgHandler(msg, handlerObject, methodName);
    }

    /**
     * Install a message message handler, @see
     * ObjectAdapter.addMsgHandler()
     *
     */
    public boolean addMsgHandler(String messageClassZ,
				 String handlerClassZ,
				 String handlerMethodZ,
				 Object handlerObject) {
	DKSOverlayAddress oa =
	    dht.myDKSImpl.getDKSRef().getOverlayAddress();
	DKSMarshal m = dht.myDKSImpl.getDKSMarshal();

	return m.addMsgHandler(oa,
			       messageClassZ,
			       handlerClassZ,
			       handlerMethodZ,
			       handlerObject);
    }
}
